<?php
declare(strict_types=1);
namespace Framework;

use App\Event;
use Framework\Support\MySqlDb;
use Framework\Support\RedisCache;
use Framework\Support\SqliteDb;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\Table;
use Swoole\Server;
use Swoole\WebSocket\Server as WebSocketServer;
use Swoole\Http\Server as HttpServer;
use Swoole\WebSocket\Frame;
use Framework\Process;

class Service extends \stdClass
{
    const VERSION = '1.0.1';
    public string $name = 'BzPHP';
    public string $pidFile = '';
    private string $logFile = '';
    private array $set = ['daemonize' => false];
    public HttpServer|WebSocketServer $http_server;

    public Event $event;
    public Libs\App $app;

    public string $start_file;

    public function __construct($ip = '0.0.0.0', $set = [])
    {
        if (empty($set)) {
            $set["enable_reuse_port"] = true;
        }

        $this->set = array_merge($this->set, $set);
        error_reporting(E_ALL);

        define('__VERSION__', self::VERSION);
        define('__WNUM__', $this->set['worker_num']);
        define('__TNUM__', $this->set['task_worker_num']);

        $this->parseCommand();
        try {
            if ($this->set['open_websocket_protocol']) {
                $this->http_server = new WebSocketServer($ip, $this->set['port'], BASE);
            } else {
                $this->http_server = new HttpServer($ip, $this->set['port'], BASE);
            }
        } catch (\Exception $e) {
            exit($e->getMessage() . "\r\n");
        }
        unset($this->set['port']);



        $this->http_server->set($this->set);

        $this->http_server->on('managerstop', [$this, 'onManagerStop']);
        $this->http_server->on('managerstart', [$this, 'onManagerStart']);
        $this->http_server->on('workerstart', [$this, 'onWorkerStart']);
        $this->http_server->on('workererror', [$this, 'onWorkerError']);
        $this->http_server->on('workerexit', [$this, 'onWorkerExit']);
        $this->http_server->on('request', [$this, 'onRequest']);
        $this->http_server->on('start', [$this, 'onMasterStart']);
        if ($this->set['open_websocket_protocol']) {
            $this->http_server->on('close', [$this, 'onClose']);
            $this->http_server->on('open', [$this, 'onOpen']);
            $this->http_server->on('message', [$this, 'onMessage']);
        }
        if ($set['task_enable_coroutine']) {
            $this->http_server->on('task', [$this, 'onTaskCo']);
        } else {
            $this->http_server->on('task', [$this, 'onTask']);
        }

        $this->http_server->on('finish', [$this, 'onFinish']);
        $this->http_server->on('pipemessage', [$this, 'onPipeMessage']);

    }

    public function _T(string $name): Table
    {
        return $_ENV['Table'][$name];
    }

    public function onRequest(Request $request, Response $response): void
    {
        try {
            $this->app->run($request, $response);
        } catch (\Throwable $e) {
            //$response->status(500);
            //$response->end("Service Error");
        }
    }

    public function onPipeMessage(Server $server, int $src_worker_id, mixed $data): void
    {
        try {
            $this->app->do($data);
        } catch (\Throwable $e) {

        }
    }

    public function onTaskCo(Server $server, Server\Task $task): void
    {
        !empty($task->data['c']) && $task->data['controller'] = $task->data['c'];
        !empty($task->data['a']) && $task->data['action'] = $task->data['a'];
        if (isset($this->app) && !empty($task->data['controller'])) {
            $this->app->task($task->data);
            $task->finish($task->data['controller'] . '_' . $task->data['action']);
        } else {
            $error = 'Error ' . xn_json_encode($task->data);
            xn_log($error, 'TaskError');
            $task->finish($error);
        }
    }

    public function onTask(Server $server, int $task_id, int $src_worker_id, $data): void
    {

        if (isset($this->app) && !empty($data['controller'])) {
            $this->app->task($data);
            $server->finish($data['controller'] . '_' . $data['action']);
        } else {
            $error = 'Error ' . xn_json_encode($data);
            xn_log($error, 'TaskError');
            $server->finish($error);
        }
    }


    public function onFinish(Server $server, int $task_id, mixed $response): int
    {
        return $task_id;
    }

    public function onMessage(WebSocketServer $server, Frame $frame): void
    {
        try {
            switch ($frame->opcode) {
                case 1:
                    !empty($this->event) && $this->event->message($frame);
                    break;
                case 2://二进制数据
                    break;
                case 8://关闭帧
                    break;
                case 9://二进制数据
                    //$server->close($frame->fd);
                    break;
                case 10:
                    break;
            }
        } catch (\Throwable $e) {
            var_dump($e->getMessage());
        }

    }

    public function onClose(WebSocketServer $server, int $fd, int $reactorId): void
    {
        try {
            !empty($this->event) && $this->event->closeClient($fd);
        } catch (\Throwable $e) {
            var_dump($e->getMessage());
        }
    }

    public function onOpen(WebSocketServer $server, Request $request): void
    {
        try {
            !empty($this->event) &&  $this->event->Open($request);
        } catch (\Throwable $e) {
            var_dump($e->getMessage());
            $server->close($request->fd);
        }
    }


    public function Table(): void
    {
        //仅供支持IDE从模型获取
        $_ENV['Table'] = [];
        $workers = new Table(2 ** 10);
        $workers->column('Request', Table::TYPE_INT);
        $workers->column('LastRequest', Table::TYPE_INT);
        $workers->column('Memory', Table::TYPE_INT);
        $workers->create();
        $_ENV['Table']['WorkersStat'] = $workers;

        $workers = new Table(2 ** 10);
        $workers->column('Request', Table::TYPE_INT);
        $workers->column('Start', Table::TYPE_INT);
        $workers->column('Time', Table::TYPE_INT);
        $workers->column('Task', Table::TYPE_INT);
        $workers->column('Pid', Table::TYPE_INT);
        $workers->create();
        $_ENV['Table']['WorkersStatus'] = $workers;

        $workers = new Table(2 ** 12);
        $workers->column('Json', Table::TYPE_INT);
        $workers->column('Value', Table::TYPE_STRING, 1024);
        $workers->create();
        $_ENV['Table']['Config'] = $workers;

        $workers = new Table(2 ** 16);
        $workers->column('Min', Table::TYPE_INT);
        $workers->column('Max', Table::TYPE_INT);
        $workers->column('Ms', Table::TYPE_INT);
        $workers->column('Count', Table::TYPE_INT);
        $workers->create();
        $_ENV['Table']['RequestStat'] = $workers;

        $workers = new Table(2 ** 14);
        $workers->column('Ms', Table::TYPE_INT);
        $workers->column('Count', Table::TYPE_INT);
        $workers->column('Sql', Table::TYPE_STRING, 1024);
        $workers->create();
        $_ENV['Table']['SqlStat'] = $workers;

        $workers = new Table(2 ** 16);
        $workers->column('Status', Table::TYPE_INT);
        $workers->create();
        $_ENV['Table']['Auth'] = $workers;

        $tables = include _include(__SYSDIR__ . 'Tables.php');
        foreach ($tables as $k => $v) {
            $workers = new Table(2 ** min(25, $v['Num']));
            foreach ($v['Fields'] as $_k => $_v) {
                if (!empty($_v[1])) {
                    $workers->column($_k, $_v[0], $_v[1]);
                } else {
                    $workers->column($_k, $_v[0]);
                }
            }
            $workers->create();
           $_ENV['Table'][$k]= $workers;
        }
        $_ENV['conf'] = include _include(__SERDIR__ . __LV__ . '_config.php');
        $_ENV['database'] = include __SERDIR__ . __LV__ . '_database.php';

        $str = [];
        if ($_ENV['conf']['debug']) {
            $_ENV['_models_name']=[];
            $include_model_files = glob(__APPDIR__ . 'Dao/*Dao.php');
            foreach ($include_model_files as $model_files) {
                $name = str_replace([__APPDIR__ . 'Dao/', 'Dao.php'], '', $model_files);
                $str[] = " * @property \Dao\\{$name}Dao \$$name";
                $_ENV['_models_name'][$name]=1;
            }
            $include_model_files = glob(__APPDIR__ . 'Dao/*/*Dao.php');
            foreach ($include_model_files as $model_files) {
                $name = str_replace([dirname($model_files) . '/', 'Dao.php'], '', $model_files);
                $str[] = " * @property \Dao\\{$name}Dao \$$name";
                $_ENV['_models_name'][$name]=1;
            }

            //从开启的插件中获取model
            $model_name = plugin_get_model();

            foreach ($model_name as $name) {
                $str[] = " * @property \Dao\\{$name}Dao \${$name}";
                $_ENV['_models_name'][$name]=1;
            }

            $str = implode("\n", $str);

            IDE_include(__CFRDIR__ . 'Libs/Controller.php', $str);
            IDE_include(__CFRDIR__ . 'Libs/Model.php', $str);
            IDE_include(__CFRDIR__ . 'Libs/Event.php', $str);
            //以上为IDE提供支持
        }

        //此处不留hook--避免麻烦 还是直接 通过Tables_middle.php 来增加吧


        //因为跨进程不共享，丢进内存目录当内存表用吧

        try {
            //is_file(__TMPDIR__ . 'ws.db') && unlink(__TMPDIR__ . 'ws.db');
            //is_file(__TMPDIR__.'tb_one_task_key.db') && unlink(__TMPDIR__.'tb_one_task_key.db');
            $sqlite = new \PDO('sqlite:' . __TMPDIR__ . 'ws.db');
            $sqlite->exec('DROP TABLE IF EXISTS ws');
            $sqlite->exec('PRAGMA journal_mode = WAL');
            $sqlite->exec('PRAGMA synchronous = OFF');
            $sqlite->exec('CREATE TABLE IF NOT EXISTS "ws" (  "uid" int NOT NULL DEFAULT 0,  "fd" int NOT NULL DEFAULT 0,  "gid" int NOT NULL DEFAULT 0,  PRIMARY KEY ("uid", "fd", "gid"))');
            $sqlite = null;
            $sqlite = new \PDO('sqlite:' . __TMPDIR__ . 'tb_one_task_key.db');
            //$sqlite->exec('DROP TABLE tb_one_task_key');
            $sqlite->exec('PRAGMA journal_mode = WAL');
            $sqlite->exec('PRAGMA synchronous = OFF');
            $sqlite->exec('PRAGMA vacuum');
            //$sqlite->exec('CREATE TABLE IF NOT EXISTS "tb_one_task_key" (  "id" integer NOT NULL, "nick_id" integer,  "module" text,  "action" TEXT,  "type" integer,  "event_key" TEXT,  "has_error" integer,  "status" integer,  "create_at" integer,  "do_num" integer,  "task_num" integer,  "async_campaign" integer, "async_wordpackage" integer,  "async_adgroup" integer,  "async_keyword" integer,  "async_crowd" TEXT,  "async_creative" TEXT,  "parent_event_key" TEXT,  "async_adzone" integer,  PRIMARY KEY ("id"));');
            $sqlite = null;
            //$sqlite->close();
        } catch (\Throwable $e) {
            var_dump($e);
        }

    }


    protected function parseCommand(): void
    {
        global $argv;
        $this->start_file = $start_file = $argv[0];
        $this->pidFile = __PIDDIR__ . $this->start_file . ".pid";
        $this->set['log_file'] = $this->logFile = __CAHDIR__ . 'Service/' . date('Ymd') . '.log';
        if (!isset($argv[1])) {
            exit("Usage: php yourfile.php {start|stop|restart|reload}\n");
        }
        $command1 = trim($argv[1]);
        $command2 = $argv[2] ?? '';
        $mode = '';
        if ($command1 === 'start') {

            if ($command2 === '-d') {
                $mode = 'in DAEMON mode';
            } else {
                $mode = 'in DEBUG mode';
            }
        }

        // Get master process PID.
        $master_pid = is_file($this->pidFile) ? (int)file_get_contents($this->pidFile) : 0;
        $master_is_alive = $master_pid && posix_kill($master_pid, 0) && posix_getpid() != $master_pid;

        if ($master_is_alive) {
            if ($command1 === 'start') {
                $this->log("Service [$start_file] already running");
                exit;
            }
        } else if ($command1 !== 'start' && $command1 !== 'restart') {
            $this->log("Service [$start_file] not run");
            exit;
        }

        switch ($command1) {
            case 'start':
                $this->Table();
                if ($command2 === '-d') {
                    $this->set['daemonize'] = true;
                }
                $this->log("Service [$start_file] $command1 $mode");
                break;
            case 'restart':
                rmdir_recursive(__TMPDIR__ . 'pid/', 1);
            case 'stop':
                $this->log("Service [$start_file] is stoping ...");
                $master_pid && posix_kill($master_pid, SIGTERM);
                $timeout = 60;
                $start_time = time();
                // Check master process is still alive?
                while (1) {
                    $master_is_alive = $master_pid && posix_kill($master_pid, 0);
                    if ($master_is_alive) {
                        // Timeout?
                        if (time() - $start_time >= $timeout) {
                            $this->log("Service [$start_file] stop fail");
                            exit;
                        }
                        // Waiting amoment.
                        usleep(1000);
                        continue;
                    }
                    // Stop success.
                    $this->log("Service [$start_file] stop success");
                    if ($command1 === 'stop') {
                        exit(0);
                    }
                    $this->set['daemonize'] = true;
                    break;
                }
                break;
            case 'reload':
                if ($command2 === '-p') {
                    rmdir_recursive(__TMPDIR__ . 'pid/', 1);
                    self::log("Service ZxProcess reload success");
                } else {
                    if ($command2 === '-g') {
                        $sig = SIGQUIT;
                    } else {
                        $sig = SIGUSR1;
                    }
                    posix_kill($master_pid, $sig);
                    self::log("Service [$start_file] reload success");
                }
                exit;
            default :
                exit("Usage: php yourfile.php {start|stop|restart|reload}\n");
        }
        (PHP_OS != 'Darwin' && !str_starts_with(PHP_OS, "CYGWIN")) && swoole_set_process_name("{$this->name} php {$this->start_file} master process");
    }

    public function onMasterStart(Server $serv): void
    {

        if (false === @file_put_contents($this->pidFile, $serv->master_pid)) {
            throw new \Exception('can not save pid to ' . $this->pidFile);
        }

        (PHP_OS != 'Darwin' && !str_starts_with(PHP_OS, "CYGWIN")) && swoole_set_process_name("{$this->name} php {$this->start_file} master process");
        //var_dump($serv->worker_pid);

    }

    public function onManagerStop(Server $serv): void
    {
        // hook root_server_onManagerStop.php
        unlink($this->pidFile);
    }

    public function onManagerStart(Server $server): void
    {

        // hook root_server_onManagerStart.php
        (PHP_OS != 'Darwin' && !str_starts_with(PHP_OS, "CYGWIN")) && swoole_set_process_name("{$this->name} php {$this->start_file} manage process");

        echo "-------------------------- ZxPHP --------------------------------\r\n";
        echo 'ZxPHP:' . self::VERSION . '  Swoole:' . SWOOLE_VERSION . '  PHP:' . PHP_VERSION . "  OPcache:" . (function_exists('opcache_reset') ? 'Open' : 'Close') . "  Run:" . __LV__ . "\r\n";
        echo "-------------------------- Table --------------------------------\r\n";
        $memorySize = 0;
        foreach ($_ENV['Table'] as $name => $table) {
            $memorySize = bcadd($memorySize.'', $table->memorySize.'');
            echo 'Name:' . str_pad($name.'', 20) . ' Size:' . str_pad($table->size.'', 10) . '  MemorySize:' . str_pad(bcdiv($table->memorySize.'', '1048576', 2) . "M", 12) . "\r\n";
        }

        echo "MemorySize:" . str_pad(bcdiv($memorySize, '1048576', 2) . 'M', 12) . "  Use: _T(Name)->Func()\r\n";
        echo "-----------------------------------------------------------------\r\n";
        echo "Master pid:" . str_pad($server->getMasterPid().'', 20) . ' Start at:' . date('Y-m-d H:i:s') . "\r\n";
        echo "Worker Num:{$this->set['worker_num']}  Task Num:{$this->set['task_worker_num']}\r\n";
        echo 'Worker Co:' . ($this->set['enable_coroutine'] ? 'true' : 'false') . '  Task Co:' . (!empty($this->set['task_enable_coroutine']) ? 'true' : 'false') . "  Max Co:{$this->set['max_coroutine']}\r\n";
        echo "Protocol HTTP:" . ($this->set['open_http_protocol'] ? 'true' : 'false') . "  WEBSOCKET:" . ($this->set['open_websocket_protocol'] ? 'true' : 'false') . "  MQTT:" . ($this->set['open_mqtt_protocol'] ? 'true' : 'false') . "   Listen Port:{$this->http_server->port}\r\n";
        echo "-------------------------- Worker --------------------------------\r\n";

    }

    public function onWorkerStart(Server $serv, int $worker_id): void
    {
        //var_dump($serv->worker_pid);
        // hook root_server_onWorkerStart.php

        if ($serv->taskworker) {
            (PHP_OS != 'Darwin' && !str_starts_with(PHP_OS, "CYGWIN")) && swoole_set_process_name("{$this->name} php {$this->start_file} tasks worker {$worker_id}");
        } else {
            (PHP_OS != 'Darwin' && !str_starts_with(PHP_OS, "CYGWIN")) && swoole_set_process_name("{$this->name} php {$this->start_file} event worker {$worker_id}");
        }

        if (function_exists('opcache_reset')) {
            opcache_reset();
        }

        $_ENV['startmemory'] = getum();
        $startime = get_ut();


        $_ENV['database'] = include __SERDIR__ . __LV__ . '_database.php';

        if ($_ENV['database']['CACHE']) {
            foreach ($_ENV['database']['CACHE'] as $name => $row) {
                $_ENV['cache_class'][$name] = new RedisCache($row,$name);
                if($worker_id==0){
                    //默认清空当前库，如果有独立说明。则默认不做处理
                    if(isset($row['no_flush'])&&!empty($row['no_flush'])){
                        continue;
                    }
                    $_ENV['cache_class'][$name]->flushDb();
                }
            }
        }

        if ($_ENV['database']['DB']) {
            foreach ($_ENV['database']['DB'] as $name => $row) {
                $_ENV['db_class'][$name] = new MySqlDb($row,$name);
            }
        }


        try {
            $_ENV['db_ws'] = new SqliteDb(__TMPDIR__ . 'ws.db');
            $_ENV['db_tb_one_task_key'] = new SqliteDb(__TMPDIR__ . 'tb_one_task_key.db');
        } catch (\Throwable $e) {
            var_dump($e);
        }
        $_ENV['plugin_srcfiles'] = [];
        $_ENV['plugin_paths'] = [];
        $_ENV['plugins'] = [];// 跟官方插件合并
        $_ENV['official_plugins'] = [];// 官方插件列表
        $_ENV['g_include_slot_kv'] = [];// 官方插件列表

        plugin_init();//插件初始化
        plugin_get_hook();//获取启用的钩子
        plugin_get_autoload();
        plugin_get_model();
        plugin_get_controller();

        // hook root_server_onWorkerStart_Common.php

        require_once _include(__COMDIR__ . 'Common.php');
        ConfigReload();
        IncludeFiles();
        // hook root_server_onWorkerStart_Tick.php

        require_once _include(__COMDIR__ . 'Tick.php');

        $this->event = new Event();
        $this->app = new Libs\App();

        echo "Worker ID:" . str_pad($worker_id.'', 4) . ' Pid:' . str_pad($serv->worker_pid.'', 8) . " Ut:" . str_pad(ut($startime), 12) . " Um:" . str_pad(um($_ENV['startmemory']), 12) . ' Task:' . ($serv->taskworker ? 'true' : 'false') . "\r\n";
    }

    public function onWorkerError(Server $server, int $worker_id, int $worker_pid, string|int $exit_code): void
    {
        $this->log("worker_pid: " . $worker_pid . " exit_code:" . $exit_code . ' ' . $server->getLastError());
        $last = $this->_T('WorkersStat')->get($worker_id.'');
        $d = $server->stats();
        $task = $server->taskworker ? 'M' : 'W';
        $workerIdk = $worker_id . $task;
        if ($last != false) {
            empty($d['worker_request_count']) and $d['worker_request_count'] = 0;
            $this->_T('WorkersStat')->set($workerIdk, ['Request' => $last['Request'] + ($d['worker_request_count'] - $last['LastRequest']), 'LastRequest' => 0, 'Memory' => memory_get_usage()]);
        } else {
            $this->_T('WorkersStat')->set($workerIdk, ['Request' => 0, 'LastRequest' => 0, 'Memory' => memory_get_usage()]);
        }
    }


    public function onWorkerExit(Server $server, int $worker_id)
    {
        echo 'WorkerExit ' . $worker_id . "\r\n";
    }

    public function log(string $msg, string $file = 'server_log'): void
    {
        xn_log($msg, $file);
        if (empty($this->set['daemonize'])) {
            echo date('Y-m-d H:i:s') . " " . $msg . "\n";
        }
    }

    public function run(): void
    {

        // hook root_server_run_start.php

        $process = new \Swoole\Process(function ($process) {


            $TWorkersStatus = $this->_T('WorkersStatus');
            $TWorkersStat = $this->_T('WorkersStat');
            $TRequestStat = $this->_T('RequestStat');
            $Current = $this->_T('Current');
            $TWorkersStatus->set('-1', ['Start' => time(), 'Task' => 2, 'Pid' => $process->pid]);
            \Swoole\Timer::tick(1000, function ($id) use($TWorkersStatus) {
                $TWorkersStatus->set('-1', ['Time' => time()]);
            });

            \Swoole\Timer::tick(1000, function ($id) use($TWorkersStatus,$TWorkersStat,$TRequestStat) {
                $TWorkersStat->incr('-1', 'Request', 1);
                $time = time();
                if ($time % 300 == 0) {
                    $works = [];
                    $request = [];
                    foreach ($TWorkersStat as $k => $row) {
                        $works[$k] = $row;
                    }
                    foreach ($TRequestStat as $k => $row) {
                        $request[$k] = $row;
                    }
                    foreach ($works as $k => $row) {
                        $TWorkersStat->del($k.'');
                        stat_log([$k, $row['Request'] < 0 ? $row['Request'] * -1 : $row['Request'], $row['LastRequest'], $row['Memory']], 'workers');
                    }
                    foreach ($request as $k => $row) {
                        $TRequestStat->set($k.'', ['Count' => 0, 'Ms' => 0]);
                        $TRequestStat->del($k.'');
                        stat_log([$k, $row['Count'], $row['Ms']], 'api');
                    }
                }
            });

            \Swoole\Timer::tick(1000, function ($id) {
                $time = time();
                if ($time % 60 == 0) {
                    $v = sys_getloadavg();
                    stat_log($v, 'sys');
                }
            });

            \Swoole\Timer::tick(60000, function ($id) use($TWorkersStatus,$TWorkersStat,$Current) {
                $TWorkersStat->incr('-1', 'Request', 1);
                $_time = bcmul(get_ut().'', '1000', 0);
                foreach ($Current as $k => $row) {
                    if ($_time - $row['Time'] > 60000) {
                        $Current->set($k.'', ['Num' => 0]);
                        $row['Num'] = 0;
                    }
                    $row['Num'] == 0 && $Current->del($k.'');
                }

            });

        }, false, 0, true);

        $this->http_server->addProcess($process);

        // hook root_server_start_end.php

        $this->http_server->start();
    }
}

?>